package com.zx.cloud.security.kafka.consumer;

import com.zx.cloud.security.constants.SecurityConstants;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

/**
 * @author zhaoxuan
 * @date 2020-09-10 16:06
 **/
//@Component
@Slf4j
public class ConsumerListener {

    @KafkaListener(topics = "${spring.kafka.topic}",groupId = "test")
    public void onMessage(ConsumerRecord<String,String> consumerRecord,Acknowledgment ack){
        log.info("onMessage消费---->topic：{},message:{},partition:{}",consumerRecord.topic(),consumerRecord.value(),consumerRecord.partition());

        ack.acknowledge();
        //System.out.println(consumerRecord.value());
    }

    @KafkaListener(topics = "${spring.kafka.topic}",groupId = "test")
    public void onMessage1(ConsumerRecord<String,String> consumerRecord,Acknowledgment ack){
        log.info("onMessage1消费---->topic：{},message:{},partition:{}",consumerRecord.topic(),consumerRecord.value(),consumerRecord.partition());
        ack.acknowledge();
        //System.out.println(consumerRecord.value());
    }
}
